package org.fastsyncer.manager.framework;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.beanutils.BeanUtils;
import org.fastsyncer.common.entity.Mapping;
import org.fastsyncer.common.entity.MappingTask;
import org.fastsyncer.worker.framework.WorkerFactory;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RouterHandle {

	private static Logger logger = LoggerFactory.getLogger(RouterHandle.class);

	private static RouterHandle instance = null;

	private RouterHandle() {
	}

	public static RouterHandle getInstance() {
		if (instance == null) {
			synchronized (RouterHandle.class) {
				instance = new RouterHandle();
			}
		}
		return instance;
	}

	/**
	 * 同步结果
	 */
	private ConcurrentHashMap<Integer, JSONObject> sync = new ConcurrentHashMap<>();

	public JSONObject router(MappingTask mappingTask, ConcurrentLinkedQueue<Integer[]> distributeTask)
			throws Exception {
		logger.info("数据分发...");

		// 目标源驱动
		List<Mapping> mappings = mappingTask.getMappings();
		int dirvers = mappings.size();
		// 线程个数
		CountDownLatch latch = new CountDownLatch(dirvers);
		// 创建线程池
		ExecutorService es = Executors.newCachedThreadPool();
		// 分配配置中设置的线程数
		for (int i = 0; i < dirvers; i++) {
			// 遍历目标源驱动
			MappingTask runTask = new MappingTask();
			BeanUtils.copyProperties(runTask, mappingTask);
			runTask.setTargetMapping(mappings.get(i));

			final Integer mapIndex = i;
			es.execute(new Runnable() {
				@Override
				public void run() {
					JSONObject handle = WorkerFactory.getInstance().handle(runTask, distributeTask);
					// 如果有错误
					if (!handle.isNull("reason")) {
						// 2.1解析错误较耗时，开个线程去存储
						new TaskLogHandle(mappingTask.getName(), handle).start();
					}
					sync.put(mapIndex, handle);
					latch.countDown();
				}
			});
		}
		// 使得主线程(main)阻塞直到latch.countDown()为零才继续执行
		latch.await();
		es.shutdown();
		logger.info("\r分发完毕...");

		// 统计分发后的执行结果
		JSONObject result = new JSONObject();
		result.put("success", 0);
		result.put("fail", 0);
		result.put("error", 0);
		for (int i = 0; i < dirvers; i++) {
			JSONObject res = sync.get(i);
			int success = result.getInt("success") + res.getInt("success");
			result.put("success", success);
			int fail = result.getInt("fail") + res.getInt("fail");
			result.put("fail", fail);
			int error = result.getInt("error") + res.getInt("error");
			result.put("error", error);
		}
		return result;
	}

}
